package com.shengzai.hdfs.text;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.*;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

/*
将如下数据上传至HDFS的/data/emp.csv中，并且使用代码将数据写入至MySQL中的emp表中
 */
public class HdfsText2 {
    public static void main(String[] args) throws IOException, ClassNotFoundException, SQLException {
        Configuration entries = new Configuration();
        entries.set("fs.defaultFS","hdfs://master:9000");
        FileSystem fileSystem = FileSystem.get(entries);
        //fileSystem.copyFromLocalFile(new Path("hadoop_code/src/data/emp.csv"),new Path("/data/emp.csv"));
        FSDataInputStream fsDataInputStream = fileSystem.open(new Path("/data/emp.csv"));
        InputStreamReader inputStreamReader = new InputStreamReader(fsDataInputStream);
        BufferedReader bufferedReader = new BufferedReader(inputStreamReader);

        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:mysql://192.168.146.136:3306/cs_db?useSSL=False&useUnicode=true&characterEncoding=UTF8", "root", "123456");
        Statement statement = con.createStatement();
        String values = "";
        String lines = null;
        while ((lines= bufferedReader.readLine())!=null){
            String[] split = lines.split(",");
            values = String.format("%s"+"('%s','%s','%s','%s','%s','%s','%s','%s'),",values,split[0],split[1],split[2],split[3],split[4],split[5],split[6],split[7]);
        }
        values = values.substring(0,values.length()-1);

        String sql = "insert into emp values"+values;
        System.out.println(sql);
        int i = statement.executeUpdate(sql);
        if (i>0){
            System.out.println("插入了"+i+"数据");
        }else {
            System.out.println("插入失败");
        }
        statement.close();
        con.close();
        bufferedReader.close();
    }
}
